package cn.com.duiba.millionaire.center.biz.mq;

import cn.com.duiba.millionaire.center.biz.mq.callback.CreditsConsumeCallback;
import cn.com.duiba.millionaire.center.biz.service.AnswerHanderService;
import cn.com.duiba.millionaire.center.biz.service.LoadReLiveCardService;
import cn.com.duiba.millionaire.center.biz.service.OpenPrizeService;
import cn.com.duiba.millionaire.center.biz.service.UnsubmitAnswerHanderService;
import cn.com.duiba.millionaire.center.biz.service.WithdrawAlipayService;
import cn.com.duiba.order.center.api.dto.CreditsCallbackMessage;
import cn.com.duiba.paycenter.message.FundTransferCallbackMessage;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.nio.charset.Charset;
import java.util.List;

/**
 * Created by zzy on 18/1/20¬.
 */
@Component("bootRocketMqMessageListener")
public class RocketMqMessageListener implements MessageListenerConcurrently {

    private static Logger logger = LoggerFactory.getLogger(RocketMqMessageListener.class);

    @Autowired
    private RocketMQTopicConstant rocketMQTopicConstant;
    @Autowired
    private AnswerHanderService answerHanderService;
    @Autowired
    private OpenPrizeService openPrizeService;
    @Autowired
    private UnsubmitAnswerHanderService  unsubmitAnswerHanderService;
    @Autowired
    private WithdrawAlipayService withdrawAlipayService;
    @Autowired
    private LoadReLiveCardService loadReLiveCardService;
    @Autowired
    private CreditsConsumeCallback creditsConsumeCallback;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        try {
            //经过duiba-spring-boot-starter的内部限制，实际上msgs里只会包含一个消息，但是不排除将来可能会改为多个，而且每个消息的topic都不同
            for (MessageExt message : msgs) {
                String topic = message.getTopic();
                byte[] body = message.getBody();
                if (topic.equals(rocketMQTopicConstant.getSubmitAnswerTopic())) {
                    answerHanderService.handle(body);
                } else if (topic.equals(rocketMQTopicConstant.getUnsubmitAnswerAnswerTopic())) {
                    unsubmitAnswerHanderService.handle(body);
                } else if (topic.equals(rocketMQTopicConstant.getOpenPrizeTopic())) {
                    String src = new String(body, Charset.forName("utf-8"));
                    if (NumberUtils.isNumber(src)) {
                        openPrizeService.openPrize(Long.valueOf(src));
                    }
                } else if(topic.equals(rocketMQTopicConstant.getWithdrawTopic())){
                    FundTransferCallbackMessage result = JSONObject.parseObject(body,FundTransferCallbackMessage.class);
                    withdrawAlipayService.withdrawAlipayCallback(result);
                } else if(topic.equals(rocketMQTopicConstant.getLoadReliveCardTopic())){
                    loadReLiveCardService.handle(body);
                } else if(rocketMQTopicConstant.getCreditsConsumeTopic().equals(topic)){
                    creditsConsumeCallback.handle(new String(body, Charset.forName("utf-8")));
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
            //发生异常时通常要重新消费，如果在发生特殊异常时你不希望该消息重新消费，则自行加个catch分支，并返回ConsumeConcurrentlyStatus.CONSUME_SUCCESS
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }

        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }

    private CreditsCallbackMessage parseMsg(byte[] msg) {
        String body;
        try {
            body = new String(msg, "utf-8");
        } catch (UnsupportedEncodingException e) {
            logger.error("UnsupportedEncodingException", e);
            return null;
        }
        return JSONObject.parseObject(body, CreditsCallbackMessage.class);
    }
}
